package portrait

import Configer.Configer
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.execution.ShuffledRowRDD
import org.apache.spark.{SparkConf, SparkContext}
import utils.GetTagsUserid

//用户画像
object DMPTags {
  def main(args: Array[String]): Unit = {
    //sparkcontext
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getName}")
    conf.setMaster("local[*]")
    conf.set("spark.serializer", Configer.serializer)
    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)
    //读取数据
    val dataFrame = sQLContext.read.parquet("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\parquet")
    val dictLines = sc.textFile("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\app_dict.txt")
    val stopLines = sc.textFile("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\stopwords.txt")

    //广播变量处理
    val dictMap: collection.Map[String, String] = dictLines.map(_.split("\t", -1)).filter(_.length >= 5).map(arr => (arr(4), arr(1))).collectAsMap()
    val dictBro = sc.broadcast(dictMap)

    val stopMap = stopLines.map((_, null)).collectAsMap()
    val stopBro = sc.broadcast(stopMap)

    //用户画像
    //存储数据
//    dataFrame.filter(GetTagsUserid.abc)
//      .map(row => {
//        //var list = List[(String, Int)]()
//        //广告位标签
//        var adList = adTags.makeTags(row)
//
//        //app名称 //渠道
//        val appList = appTags.makeTags(row, dictBro)
//
//        //设备（操作系统、联网方式、运营商）
//        val deviceList = DeviceTags.makeTags(row)
//
//        //关键字//地域标签
//        val keyList = KeyTags.makeTags(row, stopBro)
//
//        //商圈标签
//        val busList = BusinessTags.makeTags(row)


        //val userid: String = GetTagsUserid.getUserid(row)
       // (userid, adList ++ appList ++ deviceList ++ keyList ++ busList)
//      }).reduceByKey {
//      (list1, list2) =>
//        (list1 ++ list2).groupBy(_._1) //.mapValues(tp=>tp.length).toList
//          //.mapValues(tp=>tp.map(_._2).sum).toList
//          .mapValues(tp => tp.foldLeft(0)(_ + _._2)).toList
//    }.coalesce(1).saveAsTextFile("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\tags1")

    //释放资源
    sc.stop()
  }
}
